package com.jumi.microservice.rocketmq.config;


import com.alibaba.fastjson.JSONObject;
import com.jumi.microservice.common.config.rocketmq.configuration.MqConfig;
import com.jumi.microservice.mapper.AccountMapper;
import com.jumi.microservice.mapper.DepartmentMapper;
import com.jumi.microservice.mapper.LogMapper;
import com.jumi.microservice.rocketmq.listener.Mqlistener;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * - - - - - - - - - - - - - - -
 *
 * @author liuz
 * @date 2020/7/25 10:06
 * ToDoSomeThing
 * - - - - - - - - - - - - - - -
 */
@Configuration
public class MqConsumerConfig {

    /**
     * 消费者实体对象
     */
    DefaultMQPushConsumer consumer;

    @Autowired
    MqConfig mqConfig;

    @Autowired
    LogMapper logMapper;

    @Autowired
    DepartmentMapper departmentMapper;

    @Autowired
    AccountMapper accountMapper;

    /**
     * 通过构造函数 实例化对象
     */
    @Bean
    public DefaultMQPushConsumer consumer() throws MQClientException {
        System.out.println("--------------"+ JSONObject.toJSONString(mqConfig));
        consumer = new DefaultMQPushConsumer(mqConfig.getGroupName());
        consumer.setNamesrvAddr(mqConfig.getNamesrvAddr());
        //消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//        //订阅主题和 标签（ * 代表所有标签)下信息
        consumer.subscribe(mqConfig.getLogTopic(), mqConfig.getLogTag());
//        // //注册消费的监听 并在此监听中消费信息，并返回消费的状态信息
        consumer.registerMessageListener(new Mqlistener(logMapper, departmentMapper, accountMapper));
        //设置线程数
        consumer.setConsumeThreadMax(mqConfig.getThreadNum());
        consumer.start();
        System.out.println("消费者 启动成功=======");
        return consumer;
    }


}
